also catered for when category and workflow stage are empty

…date flows

subscribe_to_topic was an async generator, so its redis subscribe only ran on first iteration - after callers had already dispatched their .send() tasks.

A warmed-up worker could publish and lose completion events before anyone was listening, causing `bootstrap all` to hang waiting on events that had already been dropped. Split into open_topic_subscription (eager subscribe) and iter_topic_messages (listen loop), and updated all call sites to subscribe before sending.